package com.kitten.dataAlign.job.dataAlign;


import cn.hutool.core.collection.CollUtil;
import com.kitten.dataAlign.constants.RedisKeyConstants;
import com.kitten.dataAlign.constants.TableConstants;
import com.kitten.dataAlign.domain.mapper.DeleteMapper;
import com.kitten.dataAlign.domain.mapper.SelectMapper;
import com.kitten.dataAlign.domain.mapper.UpdateMapper;
import com.kitten.dataAlign.rpc.SearchRpcService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.List;

/**
 * @author kitten
 */
@Slf4j
@Component
public class NotePublishCountShardingXxlJob {


    @Resource
    private SelectMapper selectMapper;
    @Resource
    private UpdateMapper updateMapper;
    @Resource
    private DeleteMapper deleteMapper;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private SearchRpcService searchRpcService;

    @Value("${table.shards}")
    private int tableShards;

    @XxlJob(value = "userNotePublishCountJobHandler")
    private void userNotePublishCountJobHandler() throws Exception {
        // 分片序号
        int shardIndex = XxlJobHelper.getShardIndex();
        // 分片总数
        int shardTotal = XxlJobHelper.getShardTotal();

        String date = LocalDate.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyyMMdd"));
        String tableNameSuffix = TableConstants.buildTableNameSuffix(date, shardIndex);
        int batchSize = 1000;
        int processedCount = 0;

        for( ; ;) {
            List<Long> userIds = selectMapper.selectBatchFromDataAlignUserPublishCountTempTable(tableNameSuffix, batchSize);

            if (CollUtil.isNotEmpty(userIds)) {
                break;
            }
            userIds.forEach(userId -> {
                int publishCount = selectMapper.selectCountFromUserPublishTableByUserId(userId);

                int updateCount = updateMapper.updateUserPublishCountByUserId(userId, publishCount);
                if (updateCount > 0) {
                    String key = RedisKeyConstants.buildCountUserKey(userId);   // count:user:userId
                    Boolean hasKey = redisTemplate.hasKey(key);
                    if (hasKey) {
                        redisTemplate.opsForHash().put(key, RedisKeyConstants.FIELD_USER_NOTE_TOTAL, publishCount);
                    }
                }

            });

            // 批量物理删除这一批次记录
            deleteMapper.deleteBatchFromDataAlignUserPublishCountTempTable(tableNameSuffix, userIds);
            processedCount += userIds.size();
        }
        XxlJobHelper.log("本次处理记录数：{}", processedCount);
    }
}
